package com.fenghuaxz.rpcframework;

import com.fenghuaxz.rpcframework.annotation.LimitPolicy;
import com.fenghuaxz.rpcframework.model.MsgCallback;
import com.fenghuaxz.rpcframework.model.MsgRequest;
import com.fenghuaxz.rpcframework.model.MsgResponse;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;

import java.lang.reflect.Constructor;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 创建 : 七羽软件工作室
 * 时间 : 2017/9/28 : 1:52
 */
public class RPCEventHandler implements Runnable{

    private BaseContext context;
    private Object msg;
    private ChannelHandlerContext ctx;

    public RPCEventHandler(BaseContext context,Object msg,ChannelHandlerContext ctx){
        this.context=context;
        this.msg=msg;
        this.ctx=ctx;
    }

    @Override
    public void run() {
        try {
            if (msg instanceof MsgRequest)
                handlerRequest((MsgRequest) msg);
            else if (msg instanceof MsgResponse)
                handlerResponse((MsgResponse) msg);
        }catch (Exception e){
            ctx.fireExceptionCaught(e);
        }
    }

    private final void handlerResponse(MsgResponse msg) throws Exception{
        Map<String, MsgCallback> callbacks= this.context.callbacks;
        MsgCallback callback=callbacks.get(msg.getMsgId());
        if (callback!=null){
            callback.handler(msg,ctx.channel());
            callbacks.remove(msg.getMsgId());
        }
    }

    private final void handlerRequest(MsgRequest msg) throws Exception{
        ReentrantLock safe = context.safe;
        safe.lock();
        try {
            MsgResponse response=new MsgResponse();
            response.setMsgId(msg.getMsgId());
            try {
                SafeRemoteObject object = context.getHandler(msg.getClassName());
                if (object == null)
                    response.setException(String.format("错误:无效请求(UNREGISTER : %s)。",msg.getClassName()));
                else{
                    Class<?> targetClass = object.getClass();
                    LimitPolicy.Type[] types = new LimitPolicy.Type[2];
                    setTypes(targetClass.getAnnotation(LimitPolicy.class),types);
                    Method method = targetClass.getMethod(msg.getMethodName(),msg.getParamTypes());
                    setTypes(method.getAnnotation(LimitPolicy.class),types);
                    if (LimitPolicy.Type.ISLOCAL == types[0] && !isLocal(context,ctx.channel())){
                        String postion = msg.getClassName().concat("[").concat(method.getName().concat("]"));
                        response.setException(String.format("错误:无权限访问(ONLY %s : %s)。",LimitPolicy.Type.ISLOCAL,postion));
                        return;
                    }
                    if (LimitPolicy.Type.ISBIND == types[1] && (context.bindCheckListener != null ?
                            !context.bindCheckListener.isBind(context,ctx.channel()) : true)){
                        String postion = msg.getClassName().concat("[").concat(method.getName().concat("]"));
                        response.setException(String.format("错误:拒绝服务(ONLY %s : %s)。",LimitPolicy.Type.ISBIND,postion));
                        return;
                    }
                    object.setContext(context);
                    object.setChannel(ctx.channel());
                    response.setResult(MethodUtils.invokeMethod(object,method.getName(),msg.getParams(),msg.getParamTypes()));
                }
            }catch (Exception e){
                ctx.fireExceptionCaught(e);
                response.setException(String.valueOf(e));
            }finally {
                ctx.channel().writeAndFlush(response);
            }
        }finally {
            safe.unlock();
        }
    }

    private final void setTypes(LimitPolicy policy,LimitPolicy.Type[] types){
        LimitPolicy.Type[] find = policy != null ? policy.value() : null;
        if (find !=null)for (LimitPolicy.Type type : find){
            if (type == LimitPolicy.Type.ISLOCAL)
                types[0] = type;
            else if (type == LimitPolicy.Type.ISBIND)
                types[1] = type;
        }
    }

    private final boolean isLocal(BaseContext context,Channel channel) throws Exception{
        InetSocketAddress remote = (InetSocketAddress) channel.remoteAddress();
        InetSocketAddress local = (InetSocketAddress) channel.localAddress();
        return context.limitChannelListener != null ? context.limitChannelListener.isBelieve(context,channel)
                : remote.getHostString().equals(local.getHostString());
    }
}
